package cn.nexd.task.core;

import android.os.Handler;
import android.os.Looper;
import android.os.Message;
import android.os.Process;

import java.util.ArrayDeque;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * NexdAsyncTask create by codingfish at 15/5/25
 * TODO:
 *
 * @Version V1.0
 */
public abstract class NexdAsyncTask<Params, Progress, Result> {

    // log TAG
    private static final String TAG = "NexdAsyncTask";

    //当前设备的 CPU 核心数量
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    //线程池核心线程容量
    private static final int CORE_POOL_SIZE = CPU_COUNT + 1;

    // 线程池最大线程容量
    private static final int MAX_POOL_SIZE = CPU_COUNT * 2 + 1;

    // 线程池空闲线程的存活时间
    private static final int KEEP_ALIVE_TIME = 1;

    // priThreadFactory 线程工厂，通过工厂方法newThread来获取新线程
    private static final ThreadFactory threadFactory = new ThreadFactory() {
        private final AtomicInteger mCount = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, TAG + mCount.getAndIncrement());
        }
    };

    //静态阻塞式队列，用来存放待执行的任务，初始容量：128个
    private static final BlockingDeque<Runnable> threadPoolWorkQueue = new LinkedBlockingDeque<Runnable>(128);

    //静态并发线程池，可以用来并行执行任务
    private static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, threadPoolWorkQueue, threadFactory);

    // 任务调度器
    private static final Executor complicatingExecutor = new ComplicatingExecutor();
    private static volatile Executor serialExecutor = new SerialExecutor();

    //Handler 消息类型：发送结果
    private static final int MESSAGE_POST_RESULT = 0x1;
    //Handler 消息类型：更新进度
    private static final int MESSAGE_POST_PROGRESS = 0x2;

    // 用来发送上述两种通知，采用UI线程的Looper来处理消息.这就是为什么AsyncTask必须在UI线程调用，因为子线程默认没有Looper,无法创建下面的Handler，程序会直接Crash
    private static InternalHandler internalHandler = null;


    //任务的状态 默认为挂起，即等待执行，其类型标识为易变的（volatile）
    private volatile Status mStatus = Status.PENDING;

    //原子布尔型，支持高并发访问，标识任务是否被取消
    private final AtomicBoolean mCancelled = new AtomicBoolean();
    //原子布尔型，支持高并发访问，标识任务是否被执行过
    private final AtomicBoolean mTaskInvoked = new AtomicBoolean();

    private final WorkerRunnable<Params, Result> mWorker;
    private final FutureTask<Result> mFuture;

    private static ScheduleStrategy scheduleStrategy = ScheduleStrategy.LIFO;
    private FinishedListener finishedListener;

    public NexdAsyncTask() {

        mWorker = new WorkerRunnable<Params, Result>() {
            @Override
            public Result call() throws Exception {
                mTaskInvoked.set(true);
                android.os.Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
                return postResult(doInBackground(mParams));
            }
        };

        mFuture = new FutureTask<Result>(mWorker) {
            @Override
            protected void done() {
                try {
                    postResultIfNotInvoked(get());
                } catch (InterruptedException e) {
                    android.util.Log.w(TAG, e);
                } catch (ExecutionException e) {
                    throw new RuntimeException("An error occured while executing doInBackground()",
                            e.getCause());
                } catch (CancellationException e) {
                    postResultIfNotInvoked(null);
                }
            }
        };


        if (Looper.myLooper() != Looper.getMainLooper()) {
            internalHandler = new InternalHandler(Looper.getMainLooper());
        } else {
            internalHandler = new InternalHandler();
        }

    }

    /*该方法会在后台任务开始执行之间调用，用于进行一些界面上的初始化操作，比如显示一个进度条对话框等。*/
    protected void onPreExecute() {

    }

    /*方法中的所有代码都会在子线程中运行，我们应该在这里去处理所有的耗时任务。
    * 任务一旦完成就可以通过return语句来将任务的执行结果进行返回，如果AsyncTask的第三个泛型参数指定的是Void，就可以不返回任务执行结果。
    * 注意，在这个方法中是不可以进行UI操作的，如果需要更新UI元素，比如说反馈当前任务的执行进度，可以调用publishProgress(Progress...)方法来完成。*/
    protected abstract Result doInBackground(Params... params);

    /*当在后台任务中调用了publishProgress(Progress...)方法后，这个方法就很快会被调用，方法中携带的参数就是在后台任务中传递过来的。
    * 在这个方法中可以对UI进行操作，利用参数中的数值就可以对界面元素进行相应的更新。*/
    protected void onProgressUpdate(Progress... values) {
    }


    /*当后台任务执行完毕并通过return语句进行返回时，这个方法就很快会被调用。
    * 返回的数据会作为参数传递到此方法中，可以利用返回的数据来进行一些UI操作，比如说提醒任务执行的结果，以及关闭掉进度条对话框等。*/
    protected void onPostExecute(Result result) {

    }

    protected void onCancelled() {
    }

    protected void onCancelled(Result result) {
        onCancelled();
    }


    public enum Status {
        /**
         * 任务等待执行
         */
        PENDING,
        /**
         * 任务正在执行
         */
        RUNNING,
        /**
         * 任务已经执行结束
         */
        FINISHED
    }

    public enum ScheduleStrategy {
        /**
         * 队列中最后加入的任务最先执行
         */
        LIFO,
        /**
         * 队列中最先加入的任务最先执行
         */
        FIFO;
    }

    private void postResultIfNotInvoked(Result result) {
        final boolean wasTaskInvoked = mTaskInvoked.get();
        if (!wasTaskInvoked) {
            postResult(result);
        }
    }

    private Result postResult(Result result) {
        @SuppressWarnings("unchecked")
        Message message = internalHandler.obtainMessage(MESSAGE_POST_RESULT, new AsyncTaskResult<Result>(this, result));
        message.sendToTarget();
        return result;
    }


    public final Status getStatus() {
        return mStatus;
    }


    public final boolean isCancelled() {
        return mCancelled.get();
    }


    public final boolean cancel(boolean mayInterruptIfRunning) {
        mCancelled.set(true);
        return mFuture.cancel(mayInterruptIfRunning);
    }


    public final Result get() throws InterruptedException, ExecutionException {
        return mFuture.get();
    }


    public final Result get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return mFuture.get(timeout, unit);
    }


    public final NexdAsyncTask<Params, Progress, Result> execute(Executor executor, Params... params) {
        if (mStatus != Status.PENDING) {
            switch (mStatus) {
                case RUNNING:
                    throw new IllegalStateException("Cannot execute task:"
                            + " the task is already running.");
                case FINISHED:
                    throw new IllegalStateException("Cannot execute task:"
                            + " the task has already been executed "
                            + "(a task can be executed only once)");
            }
        }

        mStatus = Status.RUNNING;
        onPreExecute();
        mWorker.mParams = params;
        executor.execute(mFuture);

        return this;
    }

    public final NexdAsyncTask<Params, Progress, Result> executeSerial(Params... params) {
        return execute(serialExecutor, params);
    }

    public static void executeSerial(Runnable runnable) {
        serialExecutor.execute(runnable);
    }

    /**
     * 并发执行
     */
    public final NexdAsyncTask<Params, Progress, Result> executeComplicating(ScheduleStrategy scheduleStrategy, Params... params) {
        NexdAsyncTask.scheduleStrategy = scheduleStrategy;
        return execute(complicatingExecutor, params);
    }

    public static void executeComplicating(ScheduleStrategy scheduleStrategy, Runnable runnable) {
        NexdAsyncTask.scheduleStrategy = scheduleStrategy;
        complicatingExecutor.execute(runnable);
    }

    protected final void publishProgress(Progress... values) {
        if (!isCancelled()) {
            internalHandler.obtainMessage(MESSAGE_POST_PROGRESS, new AsyncTaskResult<Progress>(this, values)).sendToTarget();
        }
    }

    private void finish(Result result) {
        if (isCancelled()) {
            onCancelled(result);
            if (finishedListener != null) {
                finishedListener.onCancelled();
            }
        } else {
            onPostExecute(result);
            if (finishedListener != null) {
                finishedListener.onPostExecute();
            }
        }
        mStatus = Status.FINISHED;
    }

    private static class InternalHandler extends Handler {

        public InternalHandler(Looper mainLooper) {
            super(mainLooper);
        }

        public InternalHandler() {
            super();
        }

        @SuppressWarnings({"unchecked", "RawUseOfParameterizedType"})
        @Override
        public void handleMessage(Message msg) {
            AsyncTaskResult result = (AsyncTaskResult) msg.obj;
            switch (msg.what) {
                case MESSAGE_POST_RESULT:
                    // There is only one result
                    result.mTask.finish(result.mData[0]);
                    break;
                case MESSAGE_POST_PROGRESS:
                    result.mTask.onProgressUpdate(result.mData);
                    break;
            }
        }
    }

    private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
        Params[] mParams;
    }

    @SuppressWarnings({"RawUseOfParameterizedType"})
    private static class AsyncTaskResult<Data> {
        final NexdAsyncTask mTask;
        final Data[] mData;

        AsyncTaskResult(NexdAsyncTask task, Data... data) {
            mTask = task;
            mData = data;
        }
    }

    /**
     * 并发执行的线程调度器
     */
    private static class ComplicatingExecutor implements Executor {
        /**
         * 这里使用{@link ArrayDequeCompat}当栈比{@link java.util.Stack}性能高
         */
        private ArrayDequeCompat<Runnable> mQueue = new ArrayDequeCompat<Runnable>(complicatingMaxCount);

        /**
         * 一次同时并发的数量，根据处理器数量调节
         * <p/>
         * <p>cpu count   :  1    2    3    4    8    16    32
         * <p>once(base*2):  1    2    3    4    8    16    32
         * <p/>
         * <p>一个时间段内最多并发线程个数：
         * 双核手机：2
         * 四核手机：4
         * ...
         * 计算公式如下：
         */
        private static int complicatingOneTime;
        /**
         * 并发最大数量，当投入的任务过多大于此值时，根据Lru规则，将最老的任务移除（将得不到执行）
         * <p>cpu count   :  1    2    3    4    8    16    32
         * <p>base(cpu+3) :  4    5    6    7    11   19    35
         * <p>max(base*16):  64   80   96   112  176  304   560
         */
        private static int complicatingMaxCount;
        private static int cpuCount = CPU_COUNT;

        private void reSettings(int cpuCount) {
            this.cpuCount = cpuCount;
            complicatingOneTime = cpuCount;
            complicatingMaxCount = (cpuCount + 3) * 16;
        }

        public ComplicatingExecutor() {
            reSettings(CPU_COUNT);
        }

        @Override
        public synchronized void execute(final Runnable command) {
            Runnable r = new Runnable() {
                @Override
                public void run() {
                    command.run();
                    scheduleNext();
                }
            };

            if (threadPoolExecutor.getActiveCount() < complicatingOneTime) {
                // 小于单次并发量直接运行
                threadPoolExecutor.execute(r);
            } else {
                // 如果大于并发上限，那么移除最老的任务
                if (mQueue.size() >= complicatingMaxCount) {
                    mQueue.pollFirst();
                }
                // 新任务放在队尾
                mQueue.offerLast(r);

                // 动态获取目前cpu处理器数目,并调整设置。
                // int proCount = Runtime.getRuntime().availableProcessors();
                // if (proCount != cpuCount) {
                // cpuCount = proCount;
                // reSettings(proCount);
                // }
            }

        }

        public synchronized void scheduleNext() {
            Runnable mActive;
            switch (scheduleStrategy) {
                case LIFO:
                    mActive = mQueue.pollLast();
                    break;
                case FIFO:
                    mActive = mQueue.pollFirst();
                    break;
                default:
                    mActive = mQueue.pollLast();
                    break;
            }

            if (mActive != null) {
                threadPoolExecutor.execute(mActive);
            }
        }
    }

    private static class SerialExecutor implements Executor {
        final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();
        Runnable mActive;

        public synchronized void execute(final Runnable r) {
            mTasks.offer(new Runnable() {
                public void run() {
                    try {
                        r.run();
                    } catch (Exception e) {

                    } finally {
                        scheduleNext();
                    }
                }
            });
            if (mActive == null) {
                scheduleNext();
            }
        }

        protected synchronized void scheduleNext() {
            if ((mActive = mTasks.poll()) != null) {
                threadPoolExecutor.execute(mActive);
            }
        }
    }

    public void setFinishedListener(FinishedListener finishedListener) {
        this.finishedListener = finishedListener;
    }

    public static interface FinishedListener {
        void onCancelled();

        void onPostExecute();
    }
}
